package com.pindao.data.config;

import com.alibaba.fastjson.JSONObject;
import com.pindao.data.domain.RequestEvent;
import com.pindao.data.entity.ReqEsDocument;
import com.pindao.data.util.ElasticsearchUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Optional;


@Slf4j
@Component
public class KafkaConsumerListener {

    @Resource
    private ElasticsearchUtil elasticsearchUtil;

    @Value("${apidata.index}")
    private String index;


    @KafkaListener(topics = "${kafka.topic}")
    public void listen(List<ConsumerRecord<String, String>> records, Acknowledgment ack) {

        try {
            List<ReqEsDocument> list = new ArrayList<>();
            ReqEsDocument reqEsDocument;
            for (ConsumerRecord<String, String> record : records) {
                Optional<String> kafkaMessage = Optional.ofNullable(record.value());

                if (kafkaMessage.isPresent()) {
                    RequestEvent event = JSONObject.parseObject(kafkaMessage.get(), RequestEvent.class);
                    reqEsDocument = new ReqEsDocument();

                    reqEsDocument.setId(event.getId());
                    reqEsDocument.setResponse_body(event.getResp_body());
                    reqEsDocument.setMethod(event.getReq_method());
                    reqEsDocument.setDomain(event.getDomain());
                    reqEsDocument.setRequest_body(event.getReq_body());
                    reqEsDocument.setApi(event.getReq_uri());
                    reqEsDocument.setRequest_params(event.getReq_param());
                    reqEsDocument.setHeader(event.getHeader());

                    SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                    reqEsDocument.setTimestamp(simpleDateFormat.format(new Date()));

                    list.add(reqEsDocument);

                }
            }
            if (list.size() > 0) {
                log.info("批量提交数据:{}", list.size());
                elasticsearchUtil.insertBatch(index, list);
            }
        } catch (Exception e) {
            log.error("kafka失败，当前失败的批次。data:{} {}", records, e);
        } finally {
            ack.acknowledge();
        }

    }
}
